
set table.dynamic-table-options.enabled=true;

create function SplitFunction AS 'udtf.SplitFunction';

CREATE TABLE hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor (
    detail_msg string
)
with (
'connector' = 'kafka',
'topic' = ' hdp_lbg_supin_ods_zp_ganji_zzdetail',
'properties.bootstrap.servers' = 'bh-kafka3-1.58dns.org:9092,bh-kafka3-2.58dns.org:9092,bh-kafka3-3.58dns.org:9092,bh-kafka3-4.58dns.org:9092,bh-kafka3-5.58dns.org:9092',
'properties.group.id' = 'hdp_lbg_zhaopin-hdp_lbg_supin_ods_zp_ganji_zzdetail-dev',
'properties.client.id' = 'hdp_lbg_zhaopin-hdp_lbg_supin_ods_zp_ganji_zzdetail-RsuDe',
'scan.startup.mode' = 'group-offsets',
'sink.parallelism' = '16',
'format' = 'raw'
);

select
    msg,
    substring(msg,100,100),
    split_index(msg,'\001',2) as idx2,
    split_index(msg,'\001',3) as idx3,
    split_index(msg,'\001',37) as infoid,
    split_index(msg,'\001',46) as info_params,
    split_index(msg,'\001',34) as tjfrom
from (
  select msg
  from hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor
  , LATERAL TABLE(SplitFunction(detail_msg,'\n')) as t(msg)
)aa;



select detail_msg, substring(detail_msg, 300), split_index(detail_msg,'\001',2) as idx2
from hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor;


create function zzdetailUdtf AS 'udtf.ZpOdsZpGanjiZzdetailUdtf';


select pp.*
from (
    select regex_str
    from hdp_lbg_zhaopin_ods_zp_ganji_zzdetail_visitor
    ,LATERAL TABLE(SplitFunction(detail_msg,'\n'))
) tmp, LATERAL TABLE(zzdetailUdtf(regex_str)) as pp;


